package com.chen.config;

import com.chen.disruptor.DisruptorTemplate;
import com.chen.exception.DisruptorHandlerException;
import com.chen.model.OrderEvent;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import net.openhft.affinity.AffinityThreadFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ThreadFactory;

/**
 * 配置Disruptor 与springboot的整合
 * @Author: chen
 * @Date: 2021/5/24 2:32 PM
 * @Version 1.0
 */
@Configuration
@EnableConfigurationProperties(value = DisruptorProperties.class)
public class DisruptorAutoConfiguration {

    //加载配置文件中的信息DisruptorProperties
    public DisruptorProperties disruptorProperties ;

    public DisruptorAutoConfiguration(DisruptorProperties disruptorProperties){
        this.disruptorProperties = disruptorProperties ;
    }

    /**
     * 工厂模式去创建 OrderEvent
     * @return
     */
    @Bean
    public EventFactory<OrderEvent> eventFactory(){
        EventFactory<OrderEvent> eventFactory = new EventFactory<OrderEvent>() {
            @Override
            public OrderEvent newInstance() {
                return new OrderEvent();
            }
        };
        return eventFactory;
    }

    @Bean
    public ThreadFactory threadFactory(){
        return new AffinityThreadFactory("Match-Handler:") ;
    }

    /**
     * 设置RingBuffer等待策略  无锁高效队列
     * @return
     */
    @Bean
    public WaitStrategy waitStrategy(){
        WaitStrategy waitStrategy = new YieldingWaitStrategy();
        return waitStrategy;
    }


    @Bean
    public RingBuffer<OrderEvent> ringBuffer(EventFactory<OrderEvent> eventFactory,
                                             WaitStrategy waitStrategy,
                                             ThreadFactory threadFactory,
                                             EventHandler<OrderEvent>[] eventHandlers){
        Disruptor<OrderEvent> disruptor;

        //获取生产者对象的策略
        boolean multiProducer = disruptorProperties.isMultiProducer();

        ProducerType producerType = null;

        if(multiProducer){
            producerType = ProducerType.MULTI;
        }else {
            producerType = ProducerType.SINGLE;
        }


        disruptor = new Disruptor<OrderEvent>(eventFactory,
                disruptorProperties.getRingBufferSize(),
                threadFactory,producerType,waitStrategy);

        //添加异常处理器
        disruptor.setDefaultExceptionHandler(new DisruptorHandlerException());

        //配置事件的处理器
        disruptor.handleEventsWith(eventHandlers);

        //从disruptor对象中过去ringBuffer对象
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        // 开始监听
        disruptor.start();

        //优雅停机
        final Disruptor<OrderEvent> disruptorShutdown = disruptor ;

        // 使用优雅的停机
        Runtime.getRuntime().addShutdownHook(new Thread(
                ()->{
                    disruptorShutdown.shutdown();
                },"DisruptorShutdownThread"
        ));
        return ringBuffer ;
    }

    @Bean
    public DisruptorTemplate disruptorTemplate(RingBuffer<OrderEvent> ringBuffer){
        return new DisruptorTemplate(ringBuffer);
    }

}
